package client_agent

import (
	"encoding/json"
	"lark/conf"
	"lark/dove"
	"lark/proto"
	"lark/util"
	"net"
	"os"
	"path"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/pkg/errors"
)

type ClientProxy struct {
	sync.RWMutex
	lis            net.Listener
	status         uint32
	services       map[string]*service
	backupServices map[string]*service
	backupLock     sync.RWMutex // protect backup services
	useBackup      bool
	clientCount    uint64
	closeChan      chan struct{}
	resetEtcdCli   chan struct{}
	stopBackupChan chan struct{}
}

var _clientProxy *ClientProxy

func Init() error {
	if err := initLogger(); err != nil {
		return err
	}

	cp, err := newClientProxy()
	if err != nil {
		return err
	}

	_clientProxy = cp
	go _clientProxy.Run()
	return nil
}

//将本地client 的请求转发到远端服务器上
func newClientProxy() (*ClientProxy, error) {
	err := util.EnsureDir(conf.ServiceSnapshotPath, 0755)
	if err != nil {
		return nil, errors.WithStack(err)
	}

	strs := strings.Split(conf.ClientListenStr, "://")
	if len(strs) != 2 {
		return nil, errors.Errorf("client listen str is error:%s", conf.ClientListenStr)
	}

	network := strs[0]
	address := strs[1]

	if network == "unix" {
		err := util.EnsureDir(address, 0755)
		if err != nil {
			return nil, errors.WithStack(err)
		}
	}

	l, err := net.Listen(network, address)
	if err != nil {
		return nil, errors.WithStack(err)
	}

	if network == "unix" {
		if err := os.Chmod(address, 0766); err != nil {
			return nil, errors.WithStack(err)
		}
	}

	logger.Infof("client proxy listen on %s", conf.ClientListenStr)
	cp := &ClientProxy{lis: l, status: newFlag, closeChan: make(chan struct{}), resetEtcdCli: make(chan struct{})}

	err = cp.initService()
	if err != nil {
		return nil, err
	}
	go cp.watchUseBackup()

	return cp, nil
}

func (cp *ClientProxy) initService() error {
	etcdCli, err := newEtcdClient(conf.EtcdEndpoints)
	if err != nil {
		return errors.WithStack(err)
	}

	rev, serviceNames, addresses, workerNum, err := getEtcdServiceAddress(etcdCli)
	if err != nil {
		return errors.WithMessage(err, "get etcd service address error")
	}

	cp.setServer(serviceNames, addresses, workerNum)

	go watchEtcd(etcdCli, cp, rev)
	return nil
}

func (cp *ClientProxy) startUseBackupService() {
	cp.stopBackupChan = make(chan struct{})
begin:
	for {
		select {
		case <-cp.stopBackupChan:
		default:
			etcdCli, err := newEtcdClient(conf.BackupEtcdEndpoints)
			if err != nil {
				logger.Errorf("new backup etcd client: %s", err.Error())
				time.Sleep(time.Second)
				continue begin
			}

			rev, serviceNames, addresses, workerNum, err := getEtcdServiceAddress(etcdCli)
			if err != nil {
				logger.Errorf("get backup service addrs: %s", err.Error())
				time.Sleep(time.Second)
				continue begin
			}
			logger.Infof("start use backup etcd services")
			cp.setBackupServer(serviceNames, addresses, workerNum)
			go watchBackupEtcd(etcdCli, cp, rev)
			break begin
		}
	}

	cp.useBackup = true
}

func (cp *ClientProxy) stopUseBackupService() {
	cp.useBackup = false
	close(cp.stopBackupChan)
}

func (cp *ClientProxy) Run() {
	if !atomic.CompareAndSwapUint32(&cp.status, newFlag, runningFlag) {
		return
	}

	for {
		conn, err := cp.lis.Accept()
		if err != nil {
			if atomic.LoadUint32(&cp.status) == closedFlag {
				break
			}

			if ne, ok := err.(net.Error); ok && ne.Temporary() {
				time.Sleep(time.Millisecond * 100)
			} else {
				time.Sleep(time.Second)
			}

			logger.Errorf("accept error. %s", err)
			continue
		}

		cp.clientCount++
		id := "client-" + strconv.FormatUint(cp.clientCount, 10)
		logger.Debugf("new client %s", id)
		go newClient(cp, id, proto.NewlarkProtoV2(conn, conf.ClientAgentReadLocalTimeout(), conf.ClientAgentWriteLocalTimeout())).run()
	}

	logger.Info("client proxy exit")
}

func (cp *ClientProxy) getService(name string) (*service, error) {
	if cp.useBackup {
		return cp.getBackupService(name)
	}
	cp.RLock()
	defer cp.RUnlock()

	if _, ok := cp.services[name]; !ok {
		return nil, errors.Errorf("service not find. %s", name)
	}

	s := cp.services[name]
	return s, nil
}

func (cp *ClientProxy) getBackupService(name string) (*service, error) {
	cp.backupLock.RLock()
	defer cp.backupLock.RUnlock()

	if _, ok := cp.backupServices[name]; !ok {
		return nil, errors.Errorf("backup service not find. %s", name)
	}

	s := cp.backupServices[name]
	return s, nil
}

func (cp *ClientProxy) snapshot(name string) {
	service, ok := cp.services[name]
	if !ok {
		os.Remove(path.Join(conf.ServiceSnapshotPath, name))
		return
	}

	file, err := os.OpenFile(path.Join(conf.ServiceSnapshotPath, name), os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0666)
	if err != nil {
		logger.Fatal(err)
	}

	var str = make([]string, 0, len(service.servers))
	for _, s := range service.servers {
		str = append(str, s.raddr)
	}
	file.WriteString(strings.Join(str, ","))
	file.Close()
}

func (cp *ClientProxy) setServer(names, addresses, workerNum []string) {
	cp.Lock()
	defer cp.Unlock()

	services := make(map[string]*service)
	for i := 0; i < len(names); i++ {
		if _, ok := services[names[i]]; !ok {
			services[names[i]] = newService(names[i])
		}

		services[names[i]].addServer(addresses[i], workerNum[i])
	}

	old := cp.services
	cp.services = services

	for _, s := range old {
		s.delAll()
	}

	filepath.Walk(conf.ServiceSnapshotPath, func(p string, info os.FileInfo, err error) (returnErr error) {
		if info != nil && !info.IsDir() {
			os.Remove(p)
		}
		return
	})

	for name := range cp.services {
		cp.snapshot(name)
	}

}

func (cp *ClientProxy) setBackupServer(names, addresses, workerNum []string) {
	cp.backupLock.Lock()
	defer cp.backupLock.Unlock()

	services := make(map[string]*service)
	for i := 0; i < len(names); i++ {
		if _, ok := services[names[i]]; !ok {
			services[names[i]] = newService(names[i])
		}

		services[names[i]].addServer(addresses[i], workerNum[i])
	}

	old := cp.backupServices
	cp.backupServices = services

	for _, s := range old {
		s.delAll()
	}
}

func (cp *ClientProxy) addServer(name, address, workerNum string) {
	cp.Lock()
	defer cp.Unlock()

	if _, ok := cp.services[name]; !ok {
		cp.services[name] = newService(name)
	}

	if cp.services[name].addServer(address, workerNum) {
		cp.snapshot(name)
	}

}

func (cp *ClientProxy) addBackupServer(name, address, workerNum string) {
	cp.backupLock.Lock()
	defer cp.backupLock.Unlock()

	if _, ok := cp.backupServices[name]; !ok {
		cp.backupServices[name] = newService(name)
	}

	cp.backupServices[name].addServer(address, workerNum)
}

func (cp *ClientProxy) delServer(name, address string) {
	cp.Lock()
	defer cp.Unlock()

	if service, ok := cp.services[name]; ok {
		if service.delServer(address) {
			if service.empty() {
				delete(cp.services, name)
			}

			cp.snapshot(name)
		}
	}

}
func (cp *ClientProxy) delBackupServer(name, address string) {
	cp.backupLock.Lock()
	defer cp.backupLock.Unlock()

	if service, ok := cp.backupServices[name]; ok {
		if service.delServer(address) {
			if service.empty() {
				delete(cp.backupServices, name)
			}
		}
	}

}
func (cp *ClientProxy) ServiceList() map[string][]string {
	cp.Lock()
	defer cp.Unlock()

	var m = make(map[string][]string)

	for _, service := range cp.services {
		var ips []string
		for _, server := range service.servers {
			ips = append(ips, server.raddr)
		}
		m[service.id] = ips
	}

	return m
}

func (cp *ClientProxy) BackupServiceList() map[string][]string {
	cp.backupLock.Lock()
	defer cp.backupLock.Unlock()

	var m = make(map[string][]string)

	for _, service := range cp.backupServices {
		var ips []string
		for _, server := range service.servers {
			ips = append(ips, server.raddr)
		}
		m[service.id] = ips
	}

	return m
}

func (cp *ClientProxy) Close() error {
	if atomic.CompareAndSwapUint32(&cp.status, runningFlag, closedFlag) {
		cp.lis.Close()
	}
	return nil
}

func (cp *ClientProxy) watchUseBackup() {
	ticker := time.Tick(time.Second * 5)
	var prevBackup bool
	var bootstrap = true
	for {
		select {
		case <-ticker:
			data, err := dove.Get(conf.DoveKeyUseBackup)
			if err != nil {
				logger.Errorf("get dove config %s: %s", conf.DoveKeyUseBackup, err.Error())
				continue
			}

			var useBackup bool
			err = json.Unmarshal(data, &useBackup)
			if err != nil {
				logger.Errorf("unmarshal usebackup dove key: %s : %s", err.Error(), data)
				continue
			}
			// 第一次启动时获取到的配置就是使用备份的etcd
			if bootstrap && useBackup {
				cp.startUseBackupService()
				bootstrap = false
			} else {
				if prevBackup != useBackup {
					if useBackup {
						cp.startUseBackupService()
					} else {
						cp.stopUseBackupService()
					}
				}
			}
			prevBackup = useBackup

		}
	}
}

func ServiceList() map[string][]string {
	if _clientProxy != nil {
		return _clientProxy.ServiceList()
	}
	return nil
}

func BackupServiceList() map[string][]string {
	if _clientProxy != nil {
		return _clientProxy.BackupServiceList()
	}
	return nil
}

func ResetServiceList() error {
	select {
	case <-_clientProxy.resetEtcdCli:
		return nil
	default:
		return errors.New("reset etcd client blocked")
	}
}
